/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by Wangyunlai on 2022/07/05.
//

#include "sql/expr/expression.h"
#include "sql/expr/tuple.h"
#include "sql/operator/project_physical_operator.h"

using namespace std;

RC FieldExpr::get_value(const Tuple &tuple, Value &value) const
{
  RC rc = tuple.find_cell(TupleCellSpec(table_name(), field_name()), value);
  if (rc == RC::SUCCESS)
    return rc;
  for(int i=0;i<p_tuples_.size();i++){
    const Tuple* p_tuple=p_tuples_[i];
    if(p_tuple!=nullptr){
      rc=p_tuple->find_cell(TupleCellSpec(table_name(), field_name()), value);
      if(rc==RC::SUCCESS)return rc;
    }
  }
  return rc;
}

SubSelectExpr::SubSelectExpr(std::unique_ptr<PhysicalOperator> &&oper, const TupleSchema &schema, CompOp comp, bool is_right)
{
  operator_ = std::move(oper);
  schema_ = new TupleSchema(schema);
  comp_ = comp;
  is_right_ = is_right;
}
void SubSelectExpr::set_p_schema(const TupleSchema &p_schema)
{
  if (p_schema_ != nullptr)
    delete p_schema_;
  p_schema_ = new TupleSchema(p_schema);
}
SubSelectExpr::~SubSelectExpr()
{
  if (p_schema_ != nullptr)
    delete p_schema_;
  p_schema_ = nullptr;
  if (schema_ != nullptr)
    delete schema_;
  schema_ = nullptr;
  operator_ = nullptr;
}
RC SubSelectExpr::get_value(const Tuple &tuple, Value &value) const
{ 
  operator_->open(trx_);
  for(int i=0;i<p_tuples_.size();i++){
    const Tuple* p_tuple=p_tuples_[i];
    operator_->add_p_tuple(p_tuple);
  }
  operator_->add_p_tuple(&tuple);
  RC rc = RC::SUCCESS;
  if (comp_ == CompOp::NO_OP)
  {
   
    rc = operator_->next();
    if (rc != RC::SUCCESS)
    { 
      operator_->close();
      operator_->pop_p_tuple();
      for(int i=0;i<p_tuples_.size();i++){
        operator_->pop_p_tuple();
      }
      return rc;
    }
    Tuple *this_tuple = operator_->current_tuple();
    int cell_num = schema_->cell_num();
    std::vector<Value> values;
    values.reserve(cell_num);
    for (int i = 0; i < cell_num; i++)
    {
      const TupleCellSpec cell_spec = schema_->cell_at(i);
      Value tmp;
      rc = this_tuple->find_cell(cell_spec, tmp);
      if (rc != RC::SUCCESS)
      {
        rc = this_tuple->cell_at(i, tmp);
      }
      if (rc != RC::SUCCESS)
      {
        operator_->pop_p_tuple();
        for(int i=0;i<p_tuples_.size();i++){
        operator_->pop_p_tuple();
      }
        return rc;
      }
      values.push_back(tmp);
    }
    value.set_list(values);
      operator_->pop_p_tuple();
      for(int i=0;i<p_tuples_.size();i++){
        operator_->pop_p_tuple();
      }
    return rc;
  }
  std::vector<Value> p_values;
  if (p_schema_ != nullptr)
  {
    int cell_num = p_schema_->cell_num();
    p_values.reserve(cell_num);
    for (int i = 0; i < cell_num; i++)
    {
      const TupleCellSpec cell_spec = p_schema_->cell_at(i);
      Value tmp;
      rc = tuple.find_cell(cell_spec, tmp);
      if (rc != RC::SUCCESS)
      {
        operator_->pop_p_tuple();
        for(int i=0;i<p_tuples_.size();i++){
        operator_->pop_p_tuple();
      }
        return rc;
      }
      p_values.push_back(tmp);
    }
  }
  else
  {
    if (value.attr_type() == AttrType::LIST)
    {
      p_values = value.get_list();
    }
    else
      p_values.push_back(value);
  }
  bool result;
  
  rc = this->compare(p_values, result);
  value.set_boolean(result);
  operator_->close();
  operator_->pop_p_tuple();
  for(int i=0;i<p_tuples_.size();i++){
        operator_->pop_p_tuple();
      }
  return rc;
}
RC SubSelectExpr::compare(const std::vector<Value> &p_values, bool &result) const
{
  result = false;
  switch(comp_){
    case NOT_IN:{
      result=true;
    }break;
    case NOT_EXIST:{
      result=true;
    }break;
  }
  RC rc = RC::SUCCESS;
  std::vector<Value> values;
  int cell_num = schema_->cell_num();
  values.reserve(cell_num);
  while (true)
  {
    values.clear();
    rc = operator_->next();
    if (rc == RC::SUCCESS)
    {
      Tuple *tuple = operator_->current_tuple();
      for (int i = 0; i < cell_num; i++)
      {
        const TupleCellSpec cell_spec = schema_->cell_at(i);
        Value tmp;
        rc = tuple->find_cell(cell_spec, tmp);
        if (rc != RC::SUCCESS)
        {
          rc = tuple->cell_at(i, tmp);
        }
        if (rc != RC::SUCCESS)
        {
          result = false;
          return rc;
        }
        values.push_back(tmp);
      }
    }
    if(rc==RC::RECORD_EOF){
     rc = RC::SUCCESS;
     return rc;
    }
    if(rc!=RC::SUCCESS){
      return rc;
    }
    switch (comp_)
    {
    case EQUAL_TO:
    {

      int p_cell_num = p_values.size();
      int size = values.size();
      if (size == 0)
      {
        return rc;
      }
      RC has_next = operator_->next();
      if (has_next == RC::SUCCESS)
      {
        LOG_ERROR("multy rows in sub_select");
        return RC::INVALID_ARGUMENT;
      }
      if (p_cell_num != size)
      {
        result = false;
        return rc;
      }
      for (int i = 0; i < size; i++)
      {
        Value p_value = p_values[i];
        Value value = values[i];
        if(value.attr_type()==AttrType::NULLS||p_value.attr_type()==AttrType::NULLS){
          result=false;
          return rc;
        }
        int cmp_result = 0;
        if (is_right_)
        {
          cmp_result = p_value.compare(value);
        }
        else
        {
          cmp_result = value.compare(p_value);
        }
        result = (0 == cmp_result);
        if (result == false)
          break;
      }
      if (result == false)
        return rc;
    }
    break;
    case LESS_EQUAL:
    {
      int p_cell_num = p_values.size();
      int size = values.size();
      if (size == 0)
      {
        return rc;
      }
      RC has_next = operator_->next();
      if (has_next == RC::SUCCESS)
      {
        LOG_ERROR("multy rows in sub_select");
        return RC::INVALID_ARGUMENT;
      }
      if (p_cell_num != size)
      {
        result = false;
        return rc;
      }
      for (int i = 0; i < size; i++)
      {
        Value p_value = p_values[i];
        Value value = values[i];
        int cmp_result = 0;
        if(value.attr_type()==AttrType::NULLS||p_value.attr_type()==AttrType::NULLS){
          result=false;
          return rc;
        }
        if (is_right_)
        {
          cmp_result = p_value.compare(value);
        }
        else
        {
          cmp_result = value.compare(p_value);
        }
        result = (cmp_result <= 0);
        if (result == false)
          break;
      }
      if (result == false)
        return rc;
    }
    break;
    case NOT_EQUAL:
    {
      result = true;
      int p_cell_num = p_values.size();
      int size = values.size();
      if (size == 0)
      {
        result=false;
        return rc;
      }
      RC has_next = operator_->next();
      if (has_next == RC::SUCCESS)
      {
        LOG_ERROR("multy rows in sub_select");
        return RC::INVALID_ARGUMENT;
      }
      if (p_cell_num != size)
      {
        result = true;
        return rc;
      }
      for (int i = 0; i < size; i++)
      {
        Value p_value = p_values[i];
        Value value = values[i];
        int cmp_result = 0;
        if(value.attr_type()==AttrType::NULLS||p_value.attr_type()==AttrType::NULLS){
          result=false;
          return rc;
        }
        if (is_right_)
        {
          cmp_result = p_value.compare(value);
        }
        else
        {
          cmp_result = value.compare(p_value);
        }
        result = (cmp_result != 0);
        if (result == true)
          break;
      }
      if (result == false)
        return rc;
    }
    break;
    case LESS_THAN:
    {
      int p_cell_num = p_values.size();
      int size = values.size();
      if (size == 0)
      {
        return rc;
      }
      RC has_next = operator_->next();
      if (has_next == RC::SUCCESS)
      {
        LOG_ERROR("multy rows in sub_select");
        return RC::INVALID_ARGUMENT;
      }
      if (p_cell_num != size)
      {
        result = false;
        return rc;
      }
      for (int i = 0; i < size; i++)
      {
        Value p_value = p_values[i];
        Value value = values[i];
        int cmp_result = 0;
        if(value.attr_type()==AttrType::NULLS||p_value.attr_type()==AttrType::NULLS){
          result=false;
          return rc;
        }
        if (is_right_)
        {
          cmp_result = p_value.compare(value);
        }
        else
        {
          cmp_result = value.compare(p_value);
        }
        result = (cmp_result < 0);
        if (result == false)
          break;
      }
      if (result == false)
        return rc;
    }
    break;
    case GREAT_EQUAL:
    {
      int p_cell_num = p_values.size();
      int size = values.size();
      if (size == 0)
      {
        return rc;
      }
      RC has_next = operator_->next();
      if (has_next == RC::SUCCESS)
      {
        LOG_ERROR("multy rows in sub_select");
        return RC::INVALID_ARGUMENT;
      }
      if (p_cell_num != size)
      {
        result = false;
        return rc;
      }
      for (int i = 0; i < size; i++)
      {
        Value p_value = p_values[i];
        Value value = values[i];
        if(value.attr_type()==AttrType::NULLS||p_value.attr_type()==AttrType::NULLS){
          result=false;
          return rc;
        }
        int cmp_result = 0;
        if (is_right_)
        {
          cmp_result = p_value.compare(value);
        }
        else
        {
          cmp_result = value.compare(p_value);
        }
        result = (cmp_result >= 0);
        if (result == false)
          break;
      }
      if (result == false)
        return rc;
    }
    break;
    case GREAT_THAN:
    {
      int p_cell_num = p_values.size();
      int size = values.size();
      if (size == 0)
      {
        return rc;
      }
      RC has_next = operator_->next();
      if (has_next == RC::SUCCESS)
      {
        LOG_ERROR("multy rows in sub_select");
        return RC::INVALID_ARGUMENT;
      }
      if (p_cell_num != size)
      {
        result = false;
        return rc;
      }
      for (int i = 0; i < size; i++)
      {
        Value p_value = p_values[i];
        Value value = values[i];
        int cmp_result = 0;
        if(value.attr_type()==AttrType::NULLS||p_value.attr_type()==AttrType::NULLS){
          result=false;
          return rc;
        }
        if (is_right_)
        {
          cmp_result = p_value.compare(value);
        }
        else
        {
          cmp_result = value.compare(p_value);
        }
        result = (cmp_result > 0);
        if (result == false)
          break;
      }
      if (result == false)
        return rc;
    }
    break;
    case _IN:
    {
      if (!is_right_)
      {
        return RC::INTERNAL;
      }
      int p_cell_num = p_values.size();
      int size = values.size();
      if (size == 0 || p_cell_num != size)
      {
        result = false;
        return rc;
      }
      for (int i = 0; i < size; i++)
      {
        Value p_value = p_values[i];
        Value value = values[i];
        int cmp_result = p_value.compare(value);
        result = (0 == cmp_result);
        if (result == false)
          break;
      }
      if (result == true)
        return rc;
    }
    break;
    case NOT_IN:
    {
      if (!is_right_)
        return RC::INTERNAL;
      int p_cell_num = p_values.size();
      int size = values.size();
      if (size == 0 || p_cell_num != size)
      {
        result = true;
        return rc;
      }
      for (int i = 0; i < size; i++)
      {
        Value p_value = p_values[i];
        Value value = values[i];
        int cmp_result = p_value.compare(value);
        result = (0 != cmp_result);
        if (result == true)
          break;
      }
      if (result == false)
      {
        return rc;
      }
    }
    break;
    case _EXIST:
    {
      int size = values.size();
      if (!is_right_)
        return RC::INTERNAL;
      if (size > 0)
      {
        result = true;
      }
      return rc;
    }
    break;
    case NOT_EXIST:
    {
      int size = values.size();
      if (size == 0)
      {
        result = true;
      }
      return rc;
    }
    break;
    default:
    {
      LOG_WARN("unsupported comparison. %d", comp_);
      rc = RC::INTERNAL;
    }
    break;
    }
  }
}
RC ValueExpr::get_value(const Tuple &tuple, Value &value) const
{
  value = value_;
  return RC::SUCCESS;
}
RC FuncExpr::get_value(const Tuple &tuple, Value &value) const
{
  return try_get_value(value);
}
RC FuncExpr::compute_next(const Tuple &tuple, Value &value)
{
  RC rc = RC::SUCCESS;
  std::vector<Value> values;
  if (result_value_.attr_type() == AttrType::LIST)
  {

    for (int i = 0; i < fields_.size(); i++)
    {
      Field field = fields_[i];
      Value val;
      rc = tuple.find_cell(TupleCellSpec(field.table_name(), field.field_name()), val);
      if (rc != RC::SUCCESS)
        return rc;
      values.push_back(val);
    }
    value.set_list(values);
  }
  else
  {
    Field field = fields_[0];
    rc = tuple.find_cell(TupleCellSpec(field.table_name(), field.field_name()), value);
    values.push_back(value);
  }
  switch (func_type_)
  {
  case FuncType::F_MAX:
  {
    if (result_value_.attr_type() == LIST)
      return RC::INVALID_ARGUMENT;
    if (row_num == 0)
    {
      result_value_ = value;
    }
    if(value.attr_type()==AttrType::NULLS){
      return rc;
    }
    int compare_result = value.compare(result_value_);
    if (compare_result > 0)
    {
      result_value_ = value;
    }
  }
  break;
  case FuncType::F_MIN:
  {
    if (result_value_.attr_type() == LIST)
      return RC::INVALID_ARGUMENT;
    if (row_num == 0)
    {
      result_value_ = value;
    }
     if(value.attr_type()==AttrType::NULLS){
      return rc;
    }
    int compare_result = value.compare(result_value_);
    if (compare_result < 0)
    {
      result_value_ = value;
    }
  }
  break;
  case FuncType::F_AVG:
  {
    if (result_value_.attr_type() == LIST)
      return RC::INVALID_ARGUMENT;
    for (int i = 0; i < sums.size(); i++)
    {
      double sum = sums[i];
      switch (fields_[i].attr_type())
      {
      case AttrType::INTS:
      {
        if(values[i].attr_type()!=AttrType::NULLS){
        sum = values[i].get_int() + sum;
        }
        else {
          result_value_.set_null();
        }
      }
      break;
      case AttrType::FLOATS:
      {
        if(values[i].attr_type()!=AttrType::NULLS){
        sum = values[i].get_float() + sum;
        }
        else {
          result_value_.set_null();
        }
      }
      break;
      default:
      {
        rc = RC::INVALID_ARGUMENT;
      }
      }
      sums[i] = sum;
    }
  }
  break;
  case FuncType::F_SUM:
  {
    if (result_value_.attr_type() == LIST)
      return RC::INVALID_ARGUMENT;
    for (int i = 0; i < sums.size(); i++)
    {
      double sum = sums[i];
      switch (fields_[i].attr_type())
      {
      case AttrType::INTS:
      {
        if(values[i].attr_type()!=AttrType::NULLS){
        sum = values[i].get_int() + sum;
        }
        else {
          result_value_.set_null();
        }
      }
      break;
      case AttrType::FLOATS:
      {
       if(values[i].attr_type()!=AttrType::NULLS){
        sum = values[i].get_float() + sum;
        }
        else {
          result_value_.set_null();
        }
      }
      break;
      default:
      {
        rc = RC::INVALID_ARGUMENT;
      }
      }
      sums[i] = sum;
    }
  }
  break;
  }
  for(int i=0;i<values.size();i++){
    if(values[i].attr_type()!=AttrType::NULLS){
    row_num++;
    break;
    }
  }
  
  return rc;
}
RC FuncExpr::try_get_value(Value &value) const
{
  switch (func_type_)
  {
  case FuncType::F_MAX:
  {
    value = result_value_;
    return RC::SUCCESS;
  }
  break;
  case FuncType::F_MIN:
  {
    value = result_value_;
    return RC::SUCCESS;
  }
  break;
  case FuncType::F_COUNT:
  {
    value.set_int(row_num);
    return RC::SUCCESS;
  }
  break;
  case FuncType::F_AVG:
  {
    if (result_value_.attr_type() != LIST)
    {
      if (row_num != 0)
      {
        float avg = (float)sums[0] / row_num;
        value.set_float(avg);
      }
      else
      {
        if(result_value_.attr_type()==NULLS){
        value.set_null();
        }
        else {
          value.set_float(0);
        }
      }
    }
    else
    {
      std::vector<Value> values = result_value_.get_list();
      for (int i = 0; i < sums.size(); i++)
      {

        Value &val = values[i];
        if (row_num != 0)
        {
          float avg = (float)sums[i] / row_num;
          val.set_float(avg);
        }
        else
        {
          val.set_float(0);
        }
      }
      value.set_list(values);
    }
    return RC::SUCCESS;
  }
  break;
  case FuncType::F_SUM:
  {
    if(result_value_.attr_type()== AttrType::LIST)
  {
    std::vector<Value> values = result_value_.get_list();
    for (int i = 0; i < values.size(); i++)
    {
      Value &val = values[i];
      switch (val.attr_type())
      {
      case AttrType::INTS:
      {
        val.set_int((int)sums[i]);
        return RC::SUCCESS;
      }
      break;
      case AttrType::FLOATS:
      {
        val.set_float((float)sums[i]);
        return RC::SUCCESS;
      }
      break;
      }
    }
    value.set_list(values);
    return RC::SUCCESS;
  }
    switch (fields_[0].attr_type())
  {
    case AttrType::INTS:
    {
      if(row_num!=0){
      value.set_int((int)sums[0]);
      
      }
      else {
        if(result_value_.attr_type()==NULLS){
          value.set_null();
        }
        else {
          value.set_int(0);
        }
      }
      return RC::SUCCESS;
    }break;
    case AttrType::FLOATS:
    {
      if(row_num!=0){
      value.set_float((float)sums[0]);
      }
      else {
        if(result_value_.attr_type()==NULLS){
          value.set_null();
        }
        else {
          value.set_float(0);
        }
      }
      return RC::SUCCESS;
    } break;
  default:{
    return RC::INVALID_ARGUMENT;
    }
  }
    return RC::SUCCESS;
  }
  }
}
ValueListExpr::~ValueListExpr()
{
  if (p_schema_ != nullptr)
    delete p_schema_;
  p_schema_ = nullptr;
}
RC ValueListExpr::get_value(const Tuple &tuple, Value &value) const
{
  RC rc = RC::SUCCESS;
  std::vector<Value> p_values;
  if (p_schema_ != nullptr)
  {
    int cell_num = p_schema_->cell_num();
    p_values.reserve(cell_num);
    for (int i = 0; i < cell_num; i++)
    {
      const TupleCellSpec cell_spec = p_schema_->cell_at(i);
      Value tmp;
      rc = tuple.find_cell(cell_spec, tmp);
      if (rc != RC::SUCCESS)
      {
        return rc;
      }
      p_values.push_back(tmp);
    }
  }
  else
  {
    p_values.push_back(value);
  }
  bool result;
  rc = this->compare(p_values, result);
  value.set_boolean(result);
  return rc;
}
void ValueListExpr::set_p_schema(const TupleSchema &p_schema)
{
  if (p_schema_ != nullptr)
    delete p_schema_;
  p_schema_ = new TupleSchema(p_schema);
}
RC ValueListExpr::compare(const std::vector<Value> &p_values, bool &result) const
{
  RC rc = RC::SUCCESS;
  switch (comp_)
  {
  case EQUAL_TO:
  {
    int p_cell_num = p_values.size();
    int size = value_list_.size();
    if (size == 0 || p_cell_num != size)
    {
      result = false;
      return rc;
    }
    for (int i = 0; i < size; i++)
    {
      Value p_value = p_values[i];
      Value value = value_list_[i];
      int cmp_result = 0;
      if (is_right_)
      {
        cmp_result = p_value.compare(value);
      }
      else
      {
        cmp_result = value.compare(p_value);
      }
      result = (0 == cmp_result);
      if (result == false)
        break;
    }
    return rc;
  }
  break;
  case LESS_EQUAL:
  {
    int p_cell_num = p_values.size();
    int size = value_list_.size();
    if (size == 0 || p_cell_num != size)
    {
      result = false;
      return rc;
    }
    for (int i = 0; i < size; i++)
    {
      Value p_value = p_values[i];
      Value value = value_list_[i];
      int cmp_result = 0;
      if (is_right_)
      {
        cmp_result = p_value.compare(value);
      }
      else
      {
        cmp_result = value.compare(p_value);
      }
      result = (cmp_result <= 0);
      if (result == false)
        break;
    }
    return rc;
  }
  break;
  case NOT_EQUAL:
  {
    int p_cell_num = p_values.size();
    int size = value_list_.size();
    if(size==0){
      result =false;
      return rc;
    }
    if ( p_cell_num != size)
    {
      result = true;
      return rc;
    }
    for (int i = 0; i < size; i++)
    {
      Value p_value = p_values[i];
      Value value = value_list_[i];
      int cmp_result = 0;
      if (is_right_)
      {
        cmp_result = p_value.compare(value);
      }
      else
      {
        cmp_result = value.compare(p_value);
      }
      result = (cmp_result != 0);
      if (result == true)
        break;
    }

    return rc;
  }
  break;
  case LESS_THAN:
  {
    int p_cell_num = p_values.size();
    int size = value_list_.size();
    if (size == 0 || p_cell_num != size)
    {
      result = false;
      return rc;
    }
    for (int i = 0; i < size; i++)
    {
      Value p_value = p_values[i];
      Value value = value_list_[i];
      int cmp_result = 0;
      if (is_right_)
      {
        cmp_result = p_value.compare(value);
      }
      else
      {
        cmp_result = value.compare(p_value);
      }
      result = (cmp_result < 0);
      if (result == false)
        break;
    }

    return rc;
  }
  break;
  case GREAT_EQUAL:
  {
    int p_cell_num = p_values.size();
    int size = value_list_.size();
    if (size == 0 || p_cell_num != size)
    {
      result = false;
      return rc;
    }
    for (int i = 0; i < size; i++)
    {
      Value p_value = p_values[i];
      Value value = value_list_[i];
      int cmp_result = 0;
      if (is_right_)
      {
        cmp_result = p_value.compare(value);
      }
      else
      {
        cmp_result = value.compare(p_value);
      }
      result = (cmp_result >= 0);
      if (result == false)
        break;
    }

    return rc;
  }
  break;
  case GREAT_THAN:
  {
    int p_cell_num = p_values.size();
    int size = value_list_.size();
    if (size == 0 || p_cell_num != size)
    {
      result = false;
      return rc;
    }
    for (int i = 0; i < size; i++)
    {
      Value p_value = p_values[i];
      Value value = value_list_[i];
      int cmp_result = 0;
      if (is_right_)
      {
        cmp_result = p_value.compare(value);
      }
      else
      {
        cmp_result = value.compare(p_value);
      }
      result = (cmp_result > 0);
      if (result == false)
        break;
    }

    return rc;
  }
  break;
  case _IN:
  {
    if (!is_right_)
      return RC::INTERNAL;
    int p_cell_num = p_values.size();
    int size = value_list_.size();
    if (size == 0 || p_cell_num != 1)
    {
      result = false;
      return rc;
    }
    Value p_value = p_values[0];
    for (int i = 0; i < size; i++)
    {
      Value value = value_list_[i];
      int cmp_result = p_value.compare(value);
      result = (0 == cmp_result);
      if (result == true)
        break;
    }

    return rc;
  }
  break;
  case NOT_IN:
  {
    if (!is_right_)
      return RC::INTERNAL;
    int p_cell_num = p_values.size();
    int size = value_list_.size();
    if (size == 0 || p_cell_num != 1)
    {
      result = true;
      return rc;
    }
    Value p_value = p_values[0];
    for (int i = 0; i < size; i++)
    {
      Value value = value_list_[i];
      int cmp_result = p_value.compare(value);
      result = (0 != cmp_result);
      if (result == false)
        break;
    }

    return rc;
  }
  break;
  case _EXIST:
  {
    if (!is_right_)
      return RC::INTERNAL;
    int p_cell_num = p_values.size();
    int size = value_list_.size();
    if (size == 0 || p_cell_num != 1)
    {
      result = false;
      return rc;
    }
    Value p_value = p_values[0];
    for (int i = 0; i < size; i++)
    {
      Value value = value_list_[i];
      int cmp_result = p_value.compare(value);
      result = (0 == cmp_result);
      if (result == true)
        break;
    }

    return rc;
  }
  break;
  case NOT_EXIST:
  {
    if (!is_right_)
      return RC::INTERNAL;
    int p_cell_num = p_values.size();
    int size = value_list_.size();
    if (size == 0 || p_cell_num != 1)
    {
      result = true;
      return rc;
    }
    Value p_value = p_values[0];
    for (int i = 0; i < size; i++)
    {
      Value value = value_list_[i];
      int cmp_result = p_value.compare(value);
      result = (0 != cmp_result);
      if (result == false)
        break;
    }

    return rc;
  }
  break;
  default:
  {
    LOG_WARN("unsupported comparison. %d", comp_);
    rc = RC::INTERNAL;
  }
  break;
  }
  return rc;
}
/////////////////////////////////////////////////////////////////////////////////
CastExpr::CastExpr(unique_ptr<Expression> child, AttrType cast_type)
    : child_(std::move(child)), cast_type_(cast_type)
{
}

CastExpr::~CastExpr()
{
}

RC CastExpr::cast(const Value &value, Value &cast_value) const
{
  RC rc = RC::SUCCESS;
  if (this->value_type() == value.attr_type())
  {
    cast_value = value;
    return rc;
  }

  switch (cast_type_)
  {
  case BOOLEANS:
  {
    bool val = value.get_boolean();
    cast_value.set_boolean(val);
  }
  break;
  default:
  {
    rc = RC::INTERNAL;
    LOG_WARN("unsupported convert from type %d to %d", child_->value_type(), cast_type_);
  }
  }
  return rc;
}

RC CastExpr::get_value(const Tuple &tuple, Value &cell) const
{
  RC rc = child_->get_value(tuple, cell);
  if (rc != RC::SUCCESS)
  {
    return rc;
  }

  return cast(cell, cell);
}

RC CastExpr::try_get_value(Value &value) const
{
  RC rc = child_->try_get_value(value);
  if (rc != RC::SUCCESS)
  {
    return rc;
  }

  return cast(value, value);
}

////////////////////////////////////////////////////////////////////////////////

ComparisonExpr::ComparisonExpr(CompOp comp, unique_ptr<Expression> left, unique_ptr<Expression> right)
    : comp_(comp), left_(std::move(left)), right_(std::move(right))
{
}

ComparisonExpr::~ComparisonExpr()
{
}

RC ComparisonExpr::compare_value(const Value &left, const Value &right, bool &result) const
{
  RC rc = RC::SUCCESS;
  //TODONULL
  if(left.attr_type()==AttrType::NULLS){
    if(comp_==_IS){
      result=right.attr_type()==AttrType::NULLS;
    }
    else if(comp_==IS_NOT){
      result=!(right.attr_type()==AttrType::NULLS);
    }
    else {
      result=false;
    }
    return rc;
  }
  if(right.attr_type()==AttrType::NULLS){
     if(comp_==IS_NOT){
      result=true;
    }
    else {
      result=false;
    }
    return rc;
  }
  if (comp_ == LIKE)
  {
    result = left.like(right);
    return rc;
  }
  if (comp_ == NOT_LIKE)
  {
    result = !left.like(right);
    return rc;
  }
  int cmp_result = left.compare(right);
  result = false;
  switch (comp_)
  {
  case EQUAL_TO:
  {
    result = (0 == cmp_result);
  }
  break;
  case _IS:
  {
    if(left.attr_type()==right.attr_type()){
    result=(0==cmp_result);
    }
    else {
      result=false;
    }
  }break;
  case IS_NOT:{
    if(left.attr_type()==right.attr_type()){
    result=(cmp_result!=0);
    }
    else {
      result=true;
    }
  }break;
  case LESS_EQUAL:
  {
    result = (cmp_result <= 0);
  }
  break;
  case NOT_EQUAL:
  {
    result = (cmp_result != 0);
  }
  break;
  case LESS_THAN:
  {
    result = (cmp_result < 0);
  }
  break;
  case GREAT_EQUAL:
  {
    result = (cmp_result >= 0);
  }
  break;
  case GREAT_THAN:
  {
    result = (cmp_result > 0);
  }
  break;
  default:
  {
    LOG_WARN("unsupported comparison. %d", comp_);
    rc = RC::INTERNAL;
  }
  break;
  }

  return rc;
}

RC ComparisonExpr::try_get_value(Value &cell) const
{
  if (left_->type() == ExprType::VALUE && right_->type() == ExprType::VALUE)
  {
    ValueExpr *left_value_expr = static_cast<ValueExpr *>(left_.get());
    ValueExpr *right_value_expr = static_cast<ValueExpr *>(right_.get());
    const Value &left_cell = left_value_expr->get_value();
    const Value &right_cell = right_value_expr->get_value();

    bool value = false;
    RC rc = compare_value(left_cell, right_cell, value);
    if (rc != RC::SUCCESS)
    {
      LOG_WARN("failed to compare tuple cells. rc=%s", strrc(rc));
    }
    else
    {
      cell.set_boolean(value);
    }
    return rc;
  }

  return RC::INVALID_ARGUMENT;
}

RC ComparisonExpr::get_value(const Tuple &tuple, Value &value) const
{
  Value left_value;
  Value right_value;
  if(left_->type()==ExprType::SUBSELECT&&right_->type()==ExprType::SUBSELECT ){
    RC rc=RC::SUCCESS;
    rc=left_->get_value(tuple,left_value);
    if(rc==RC::SUCCESS){
      rc=left_->get_value(tuple,left_value);
      if(rc==RC::SUCCESS){
        LOG_ERROR("multy rows in sub_select");
        return RC::INVALID_ARGUMENT;
      }
    }
    int left_cell_num=left_value.get_list().size();
    switch (comp_){
      case CompOp::_IN:{
        while(true){
          rc=right_->get_value(tuple,right_value);
          if(rc==RC::RECORD_EOF){
            value.set_boolean(false);
            rc=RC::SUCCESS;
            break;
          }
          if(rc!=RC::SUCCESS){
            return rc;
          }
          int compare_result=left_value.compare(right_value);
          if(compare_result==0){
            value.set_boolean(true);
            return rc;
          }
        }
      }break;
      case CompOp::NOT_IN:{
        while(true){
        rc=right_->get_value(tuple,right_value);
          if(rc==RC::RECORD_EOF){
            value.set_boolean(true);
            rc=RC::SUCCESS;
            break;
          }
          if(rc!=RC::SUCCESS){
            return rc;
          }
          int compare_result=left_value.compare(right_value);
          if(compare_result==0){
            value.set_boolean(false);
            return rc;
          }
        }
      }break;
      case CompOp::_EXIST:{
          while(true){
          rc=right_->get_value(tuple,right_value);
          if(rc==RC::RECORD_EOF){
            value.set_boolean(false);
            rc=RC::SUCCESS;
            break;
          }
          if(rc!=RC::SUCCESS){
            return rc;
          }
          int compare_result=left_value.compare(right_value);
          if(compare_result==0){
            value.set_boolean(true);
            return rc;
          }
          }
      }break;
      case CompOp::NOT_EXIST:{
           while(true){
          rc=right_->get_value(tuple,right_value);
          if(rc==RC::RECORD_EOF){
            value.set_boolean(true);
            rc=RC::SUCCESS;
            break;
          }
          if(rc!=RC::SUCCESS){
            return rc;
          }
          int compare_result=left_value.compare(right_value);
          if(compare_result==0){
            value.set_boolean(false);
            return rc;
          }
        }
      }break;
      default:{
        rc=right_->get_value(tuple,right_value);
        if(rc==RC::SUCCESS){
          rc=right_->get_value(tuple,right_value);
          if(rc==RC::SUCCESS){
            LOG_ERROR("multy rows in sub_select");
            return RC::INVALID_ARGUMENT;
          }
          if(rc==RC::RECORD_EOF){
            rc=RC::SUCCESS;
          }
          int right_cell_num=right_value.get_list().size();
          if(left_cell_num==0||right_cell_num==0){
            value.set_boolean(false);
            return rc;
          }
          if(left_cell_num!=right_cell_num){
            if(comp_==CompOp::NOT_EQUAL){
              value.set_boolean(true);
            }
            else {
            value.set_boolean(false);
            }
            return rc;
          }
          bool  result=false;
          rc=compare_value(left_value,right_value,result);
          if(rc==RC::SUCCESS){
            value.set_boolean(result);
          }
          else {
            value.set_boolean(false);
          }
          return rc;
        }
      }
    }
    return rc;
  }
  if (left_->type() == ExprType::SUBSELECT || left_->type() == ExprType::VALUE_LIST)
  {
    RC rc = right_->get_value(tuple, right_value);
    if (rc != RC::SUCCESS)
    {
      LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
      return rc;
    }
    rc = left_->get_value(tuple, right_value);
    value.set_boolean(right_value.get_boolean());
    return rc;
  }
  if (right_->type() == ExprType::SUBSELECT || right_->type() == ExprType::VALUE_LIST)
  {
    RC rc = left_->get_value(tuple, left_value);
    if (rc != RC::SUCCESS)
    {
      LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
      return rc;
    }
    rc = right_->get_value(tuple, left_value);
    value.set_boolean(left_value.get_boolean());
    return rc;
  }
  RC rc = left_->get_value(tuple, left_value);
  if (rc != RC::SUCCESS)
  {
    LOG_WARN("failed to get value of left expression. rc=%s", strrc(rc));
    return rc;
  }
  rc = right_->get_value(tuple, right_value);
  if (rc != RC::SUCCESS)
  {
    LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
    return rc;
  }

  bool bool_value = false;
  rc = compare_value(left_value, right_value, bool_value);
  if (rc == RC::SUCCESS)
  {
    value.set_boolean(bool_value);
  }
  return rc;
}

////////////////////////////////////////////////////////////////////////////////
ConjunctionExpr::ConjunctionExpr(CombinationType type, vector<unique_ptr<Expression>> &children)
    : conjunction_type_(type), children_(std::move(children))
{
}

RC ConjunctionExpr::get_value(const Tuple &tuple, Value &value) const
{
  RC rc = RC::SUCCESS;
  if (children_.empty())
  {
    value.set_boolean(true);
    return rc;
  }

  Value tmp_value;
  for (const unique_ptr<Expression> &expr : children_)
  {
    rc = expr->get_value(tuple, tmp_value);
    if (rc != RC::SUCCESS)
    {
      LOG_WARN("failed to get value by child expression. rc=%s", strrc(rc));
      return rc;
    }
    bool bool_value = tmp_value.get_boolean();
    if ((conjunction_type_ == CombinationType::C_AND && !bool_value) || (conjunction_type_ == CombinationType::C_OR && bool_value))
    {
      value.set_boolean(bool_value);
      return rc;
    }
  }

  bool default_value = (conjunction_type_ == CombinationType::C_AND);
  value.set_boolean(default_value);
  return rc;
}

////////////////////////////////////////////////////////////////////////////////

ArithmeticExpr::ArithmeticExpr(ArithmeticExpr::Type type, Expression *left, Expression *right)
    : arithmetic_type_(type), left_(left), right_(right)
{
}
ArithmeticExpr::ArithmeticExpr(ArithmeticExpr::Type type, unique_ptr<Expression> left, unique_ptr<Expression> right)
    : arithmetic_type_(type), left_(std::move(left)), right_(std::move(right))
{
}

AttrType ArithmeticExpr::value_type() const
{
  if (!right_)
  {
    return left_->value_type();
  }

  if (left_->value_type() == AttrType::INTS &&
      right_->value_type() == AttrType::INTS &&
      arithmetic_type_ != Type::DIV)
  {
    return AttrType::INTS;
  }

  return AttrType::FLOATS;
}

RC ArithmeticExpr::calc_value(const Value &left_value, const Value &right_value, Value &value) const
{
  RC rc = RC::SUCCESS;

  const AttrType target_type = value_type();

  switch (arithmetic_type_)
  {
  case Type::ADD:
  {
    if (target_type == AttrType::INTS)
    {
      value.set_int(left_value.get_int() + right_value.get_int());
    }
    else
    {
      value.set_float(left_value.get_float() + right_value.get_float());
    }
  }
  break;

  case Type::SUB:
  {
    if (target_type == AttrType::INTS)
    {
      value.set_int(left_value.get_int() - right_value.get_int());
    }
    else
    {
      value.set_float(left_value.get_float() - right_value.get_float());
    }
  }
  break;

  case Type::MUL:
  {
    if (target_type == AttrType::INTS)
    {
      value.set_int(left_value.get_int() * right_value.get_int());
    }
    else
    {
      value.set_float(left_value.get_float() * right_value.get_float());
    }
  }
  break;

  case Type::DIV:
  {
    if (target_type == AttrType::INTS)
    {
      if (right_value.get_int() == 0)
      {
        // NOTE: 设置为整数最大值是不正确的。通常的做法是设置为NULL，但是当前的miniob没有NULL概念，所以这里设置为整数最大值。
        value.set_int(numeric_limits<int>::max());
      }
      else
      {
        value.set_int(left_value.get_int() / right_value.get_int());
      }
    }
    else
    {
      if (right_value.get_float() > -EPSILON && right_value.get_float() < EPSILON)
      {
        // NOTE: 设置为浮点数最大值是不正确的。通常的做法是设置为NULL，但是当前的miniob没有NULL概念，所以这里设置为浮点数最大值。
        value.set_float(numeric_limits<float>::max());
      }
      else
      {
        value.set_float(left_value.get_float() / right_value.get_float());
      }
    }
  }
  break;

  case Type::NEGATIVE:
  {
    if (target_type == AttrType::INTS)
    {
      value.set_int(-left_value.get_int());
    }
    else
    {
      value.set_float(-left_value.get_float());
    }
  }
  break;

  default:
  {
    rc = RC::INTERNAL;
    LOG_WARN("unsupported arithmetic type. %d", arithmetic_type_);
  }
  break;
  }
  return rc;
}

RC ArithmeticExpr::get_value(const Tuple &tuple, Value &value) const
{
  RC rc = RC::SUCCESS;

  Value left_value;
  Value right_value;

  rc = left_->get_value(tuple, left_value);
  if (rc != RC::SUCCESS)
  {
    LOG_WARN("failed to get value of left expression. rc=%s", strrc(rc));
    return rc;
  }
  rc = right_->get_value(tuple, right_value);
  if (rc != RC::SUCCESS)
  {
    LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
    return rc;
  }
  return calc_value(left_value, right_value, value);
}

RC ArithmeticExpr::try_get_value(Value &value) const
{
  RC rc = RC::SUCCESS;

  Value left_value;
  Value right_value;

  rc = left_->try_get_value(left_value);
  if (rc != RC::SUCCESS)
  {
    LOG_WARN("failed to get value of left expression. rc=%s", strrc(rc));
    return rc;
  }

  if (right_)
  {
    rc = right_->try_get_value(right_value);
    if (rc != RC::SUCCESS)
    {
      LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
      return rc;
    }
  }

  return calc_value(left_value, right_value, value);
}